package com.atguigu.day09;

import com.atguigu.bean.WaterSensor;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Session;
import org.apache.flink.table.api.Slide;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.Tumble;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.hadoop.yarn.webapp.hamlet2.Hamlet;

import java.time.Duration;

import static org.apache.flink.table.api.Expressions.*;


public class Flink12_TableAPI_GroupWindow {
    public static void main(String[] args) throws Exception {
        //1.获取流的执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        //2.获取数据

      /*  SingleOutputStreamOperator<WaterSensor> waterSensorStream = env.fromElements(new WaterSensor("sensor_1", 1000L, 10),
                new WaterSensor("sensor_1", 2000L, 20),
                new WaterSensor("sensor_2", 3000L, 30),
                new WaterSensor("sensor_1", 4000L, 40),
                new WaterSensor("sensor_1", 5000L, 50),
                new WaterSensor("sensor_2", 6000L, 60))
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy
                                .<WaterSensor>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                                .withTimestampAssigner(new SerializableTimestampAssigner<WaterSensor>() {
                                    @Override
                                    public long extractTimestamp(WaterSensor element, long recordTimestamp) {
                                        return element.getTs();
                                    }
                                })
                );*/

        SingleOutputStreamOperator<WaterSensor> waterSensorStream = env.socketTextStream("localhost", 9999)
                .map(
                        new MapFunction<String, WaterSensor>() {
                            @Override
                            public WaterSensor map(String value) throws Exception {
                                String[] split = value.split(",");
                                return new WaterSensor(split[0], Long.parseLong(split[1]), Integer.parseInt(split[2]));
                            }
                        }
                )
                .assignTimestampsAndWatermarks(WatermarkStrategy.<WaterSensor>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                        .withTimestampAssigner(new SerializableTimestampAssigner<WaterSensor>() {
                            @Override
                            public long extractTimestamp(WaterSensor element, long recordTimestamp) {
                                return element.getTs() * 1000;
                            }
                        })
                );


        // 3.获取表的执行环境
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        // 4.将流转为表 并指定处理时间字段
        Table table = tableEnv.fromDataStream(waterSensorStream,$("id"),$("ts"),$("vc"),$("pt").proctime());
//        Table table = tableEnv.fromDataStream(waterSensorStream,$("id"),$("ts"),$("vc"),$("et").rowtime());


        table
                //TODO 5.开始基于处理时间的滚动窗口(读取无界数据流时才有效果,因为数据量太小，程序处理的时间达不到关窗的时间，所以没有窗口被触发，也就没有结果)
                .window(Tumble.over(lit(5).seconds()).on($("pt")).as("w"))
                //开启一个基于事件时间的滚动窗口 窗口大小为5s
//                .window(Tumble.over(lit(5).seconds()).on($("pt")).as("w"))
                //开启一个基于事件时间的滑动窗口 窗口大小为3s 滑动步长为2s
//                .window(Slide.over(lit(3).second()).every(lit(2).second()).on($("et")).as("w"))
                //开启一个基于事件时间的会话窗口 会话间隔为2S
//                .window(Session.withGap(lit(2).second()).on($("et")).as("w"))
                //开启一个基于元素个数的滚动窗口，窗口大小为3 只能指定处理时间字段
//                .window(Tumble.over(rowInterval(2L)).on($("pt")).as("w"))
                .groupBy($("id"),$("w"))
                .select($("id"),$("vc").sum().as("vcSum"),$("w").start().as("wStart"),$("w").end().as("wEnd"))
//                .select($("id"),$("vc").sum().as("vcSum"))
                .execute()
                .print();

    }
}
